package com.runjian.rundo.iot.agent.common.network.tcp.session;

import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Map;
import java.util.concurrent.Future;
import java.util.function.Consumer;

import com.runjian.rundo.bmp.kernel.common.spring.context.utils.MdcUtils;
import com.runjian.rundo.iot.agent.common.session.Session;

import cn.hutool.core.util.ObjUtil;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;

/**
 * Tcp会话
 * 
 * @author CaoQingHua
 */
@Slf4j
@Getter
@Setter
@AllArgsConstructor
public class TcpDeviceSession implements Session {

    /**
     * tcp管道
     */
    private Channel channel;

    @Override
    public Future<Void> writeAndFlush(ByteBuf byteBuf, Consumer<Boolean>... callbacks) {
        // 写入数据
        String byteStr = ByteBufUtil.hexDump(byteBuf);
        ChannelFuture future = this.getChannel().writeAndFlush(byteBuf);
        Map<String, String> mdcContext= MDC.getCopyOfContextMap();
        future.addListener((ChannelFutureListener)f -> {
            MDC.setContextMap(mdcContext);
            if (f.isSuccess()) {
                if (log.isDebugEnabled()) {
                    log.debug("设备数据发送成功: data = {}", byteStr);
                }

            } else {
                log.warn("设备数据发送失败: data = {}", byteStr);
            }
            // 调用回调涵数
            if (ObjUtil.isNotEmpty(callbacks)) {
                Arrays.stream(callbacks).forEach(callback -> {
                    callback.accept(f.isSuccess());
                });
            }
        });
        return future;
    }

    @Override
    public Future<Void> writeAndFlush(String data, Consumer<Boolean>... callbacks) {
        return writeAndFlush(data.getBytes(StandardCharsets.UTF_8), callbacks);
    }

    @Override
    public Future<Void> writeAndFlush(byte[] data, Consumer<Boolean>... callbacks) {
        ByteBuf byteBuf = Unpooled.buffer();
        byteBuf.writeBytes(data);
        return this.writeAndFlush(byteBuf, callbacks);
    }

    @Override
    public Boolean isActive() {
        return this.getChannel().isActive();
    }

    @Override
    public Future<Void> close(Consumer<Boolean>... callbacks) {
        ChannelFuture future = this.getChannel().close();
        future.addListener((ChannelFutureListener)f -> {
            if (log.isDebugEnabled()) {
                if (f.isSuccess()) {
                    log.debug("管道关闭成功: id = {}", this.getChannelId());
                } else {
                    log.debug("管道关闭失败: id = {}", this.getChannelId());
                }
            }
            // 调用回调涵数
            if (ObjUtil.isNotEmpty(callbacks)) {
                Arrays.stream(callbacks).forEach(callback -> {
                    callback.accept(f.isSuccess());
                });
            }
        });
        return future;
    }

    /**
     * 获取管道Id
     *
     * @return 管道Id
     */
    public String getChannelId() {
        return this.getChannel().id().asLongText();
    }

}
